package com.shujia.spark.streaming

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo6Window {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("wc")
    conf.set("spark.sql.shuffle.partitions", "1")

    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    //创建sparkStraming上下文对象,  需要指定创建RDD的间隔时间    batch时间   可以理解为多少秒计算一次
    val ssc = new StreamingContext(sc, Durations.seconds(1))

    ssc.checkpoint("spark/data/window")
    val ds = ssc.socketTextStream("node1", 8888)


    /**
      * 统计最近10秒单词的数量，每个1秒统计一次
      */

    val wordsDS = ds.flatMap(_.split(","))

    val kvDS = wordsDS.map((_, 1))


    /**
      * reduceByKeyAndWindow :统计一个window单词的数量
      * 滑动窗口
      *
      * 窗口大小和滑动时间都必须是batch的整数倍
      *
      *
      *
      */
    //    val countDS = kvDS.reduceByKeyAndWindow(
    //      (x, y) => x + y, //聚合函数
    //      windowDuration = Durations.seconds(10), //窗口大小
    //      slideDuration = Durations.seconds(1) //华东时间
    //    )


    /**
      * 优化版本
      * 需要设置checkpoint路径
      *
      * 如果窗口大小等于滑动时间（滚动窗口），就不存在重复计算的问题了，就没必要使用优化版本了
      *
      */
    val countDS = kvDS.reduceByKeyAndWindow(
      (x, y) => x + y, //聚合函数
      (i, j) => i - j, //减掉多余部分的函数
      windowDuration = Durations.seconds(10), //窗口大小
      slideDuration = Durations.seconds(1) //华东时间
    )


    countDS.filter(_._2 != 0).print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }
}
